package com.xujian.cdc.schema;

import com.alibaba.fastjson2.JSONObject;
import com.ververica.cdc.debezium.DebeziumDeserializationSchema;
import io.debezium.data.Envelope;
import lombok.extern.slf4j.Slf4j;
import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.util.Collector;
import org.apache.kafka.connect.data.Field;
import org.apache.kafka.connect.data.Struct;
import org.apache.kafka.connect.source.SourceRecord;

import java.util.Objects;


/**
 * @author star xu
 * @date 2022/12/30 11:29
 * @Description:日期格式通过转换器类处理
 * @Slogan: 致敬大师,致敬未来的自己
 */
@Slf4j
public class TxlcCustomerSchema implements DebeziumDeserializationSchema<String> {
    @Override
    public void deserialize(SourceRecord sourceRecord, Collector<String> collector) throws Exception {
        Struct value = (Struct) sourceRecord.value();
        Struct before = value.getStruct("before");
        JSONObject beforeData = transData(before);

        Struct after = value.getStruct("after");
        JSONObject afterData = transData(after);
        Envelope.Operation op = Envelope.operationFor(sourceRecord);

        log.info("->source->" + value.toString());
        log.info("===beforeData===" + beforeData.toJSONString());
        log.info(">>>afterData>>>" + afterData.toJSONString());

        if (Objects.equals(op, Envelope.Operation.DELETE)) {
            // starrocks表需要使用主键模型，另外json中需要有{"__op":1}表示删除,{"__op":0}表示upsert
            beforeData.put("__op", 1);
            collector.collect(beforeData.toJSONString());
        } else if (Objects.equals(op, Envelope.Operation.UPDATE)) {
            afterData.put("__op", 0);
        }
        collector.collect(afterData.toJSONString());

    }

    @Override
    public TypeInformation<String> getProducedType() {
        return BasicTypeInfo.STRING_TYPE_INFO;
    }

    public JSONObject transData(Struct struct){
        JSONObject result = new JSONObject();
        if (struct != null){
            for (Field field : struct.schema().fields()) {
                result.put(field.name(), struct.get(field));
            }
        }
        return result;
    }
}